package com.ry.flink.job8;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

/**
* 求每个窗口中的数据的平均值
*
* 思路：
*  累加值/总的个数=平均值
*/
public class aggregateWindowTest8 {
   public static void main(String[] args)  throws Exception{
       StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
       DataStreamSource<String> dataStream = env.socketTextStream("192.168.134.130", 9999);

       SingleOutputStreamOperator<Integer> numberStream = dataStream.map(line -> Integer.valueOf(line));
       AllWindowedStream<Integer, TimeWindow> windowStream = numberStream.timeWindowAll(Time.seconds(5));

       windowStream.aggregate(new MyAggregate()).print();
       env.execute("aggregateWindowTest");
   }

   private static class MyAggregate implements AggregateFunction<Integer,Tuple2<Integer,Integer>,Double> {

       /**
        * 累加器
        * */
       @Override
       public Tuple2<Integer, Integer> createAccumulator() {
           /**
            * key:计数
            * value:累计的值
            * */
           return new Tuple2<>(0,0);
       }

       /**
        * 针对每一条数据的操作
        * */
       @Override
       public Tuple2<Integer, Integer> add(Integer value, Tuple2<Integer, Integer> accumulator) {
           return new Tuple2<>(accumulator.f0+1,accumulator.f1+value);
       }

       @Override
       public Double getResult(Tuple2<Integer, Integer> accumulator) {
           return (double)accumulator.f1 / accumulator.f0;
       }

       @Override
       public Tuple2<Integer, Integer> merge(Tuple2<Integer, Integer> a, Tuple2<Integer, Integer> b) {
           return Tuple2.of(a.f0 + b.f0,a.f1+b.f1);
       }
   }

}
